package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.beans.TestPaperAndCourseBean;
import com.atguigu.edu.realtime.beans.TestQuestionBean;
import com.atguigu.edu.realtime.utils.ClickHouseUtil;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;

public class DwsTestPaperAndCourseWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点相关设置(略)

        // TODO 3. 从 Kafka dwd_test_question 读取问题数据
        String topic = "dwd_test_exam";
        String groupId = "dws_test_paper_course_group";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        // TODO 4. 过滤 null 数据并转换数据结构
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        if(jsonStr != null){
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        }
                    }
                }
        );



        // TODO 5. 设置水位线
        SingleOutputStreamOperator<JSONObject> withWatermarkStream = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                        return jsonObj.getLong("ts") * 1000;
                                    }
                                }
                        )
        );

        // TODO 6. 按照题目 id 分组
        KeyedStream<JSONObject, String> keyedpaperStream = withWatermarkStream.keyBy(r -> r.getString("paper_id"));
        //
        SingleOutputStreamOperator<TestPaperAndCourseBean> mapDS = keyedpaperStream.map(new MapFunction<JSONObject, TestPaperAndCourseBean>() {
            @Override
            public TestPaperAndCourseBean map(JSONObject jsonObject) throws Exception {

                String paperId = jsonObject.getString("paper_id");
                String courseId = jsonObject.getString("course_id");
                Long ts = jsonObject.getLong("ts") * 1000;
                Long usertestCount = jsonObject.getLong("user_id");
                String scoreAmount = jsonObject.getString("score");
                Long timeTestAmount = jsonObject.getLong("duration_sec");
                TestPaperAndCourseBean orderBean = TestPaperAndCourseBean.builder()
                        .paperId(paperId).usertestCount(usertestCount)
                        .courseId(courseId)
                        .scoreAmount(new BigDecimal(scoreAmount))
                        .timeTestAmount(timeTestAmount)
                        .ts(ts)
                        .build();
                return orderBean;
            }
        });

        AllWindowedStream<TestPaperAndCourseBean, TimeWindow> windowDS = mapDS.windowAll(TumblingEventTimeWindows.of(
                Time.seconds(10L)));
        SingleOutputStreamOperator<TestPaperAndCourseBean> reduceDS = windowDS.reduce(new ReduceFunction<TestPaperAndCourseBean>() {
            @Override
            public TestPaperAndCourseBean reduce(TestPaperAndCourseBean value1, TestPaperAndCourseBean value2) throws Exception {
                value1.setUsertestCount(value1.getUsertestCount() + value2.getUsertestCount());
                value1.setScoreAmount(value1.getScoreAmount().add(value2.getScoreAmount()));
                value1.setTimeTestAmount(value1.getTimeTestAmount() + value2.getTimeTestAmount());
                return value1;
            }
        }, new ProcessAllWindowFunction<TestPaperAndCourseBean, TestPaperAndCourseBean, TimeWindow>() {
            @Override
            public void process(ProcessAllWindowFunction<TestPaperAndCourseBean, TestPaperAndCourseBean, TimeWindow>.Context context, Iterable<TestPaperAndCourseBean> iterable, Collector<TestPaperAndCourseBean> collector) throws Exception {
                String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                String edt = DateFormatUtil.toYmdHms(context.window().getEnd());

                for (TestPaperAndCourseBean value : iterable) {
                    value.setStt(stt);
                    value.setEdt(edt);
                    value.setTs(System.currentTimeMillis());
                    collector.collect(value);
                }
            }
        });


        reduceDS.print(">>>>");



        //TODO 11.将聚合的结果写到CK中//DwsTestPaperAndCourseWindow
        reduceDS.addSink(
                ClickHouseUtil.getSinkFunction("insert into dws_test_paper_course_window values(?,?,?,?,?,?,?,?)")
        );
        env.execute();
    }
}



